Pythonのフレームワーク「FastAPI」をEC2で実装する

こんにちは、沖縄在住の下地です。コロナの影響でリモートワークになり早一ヶ月経ちました。家で業務をするというのは違和感はあるのですが通勤しなくて良いというのはとても魅力的だなと思い始めてます。

本日はPythonのフレームワークであるFastAPIについて興味がありEC2で実装したのでまとめたいと思います。

全体像

EC2でFastAPIを実装し、Amazon RDS(RDS)と接続し、CRUDのAPIを作成します。EC2とRDSは起動している状態からスタートします。

開発環境

環境としては以下の仕様で行います。

• python: 3.7.6
• sqlalchemy: 1.3.15
• OS: Amazon Linux 2 AMI (HVM), SSD Volume Type
• MySQL Community Edition: 5.7.22

EC2のセキュリティグループ設定

EC2でFastAPIを実装し確認するために、セキュリティグループは図に示すようにカスタムTCPタイプを選択し、ポート範囲を8000に設定します。

Python3系のインストール

EC2にpython3の環境を構築します。

$ sudo yum install python3 -y
$ python3 -m venv my_app/env
$ source my_app/env/bin/activate
$ python --version -- pythonのversion確認
Python 3.7.6

FastAPIのインストール

最新のpipモジュールがインストールされているか確認し、FastAPIとDB接続に必要なライブラリをインストールします。

$ pip install pip --upgrade
$ pip install fastapi
$ pip install uvicorn

FastAPI実装

FastAPIのDocumentを参考にmain.pyを作成します。 EC2の外からアクセスできるように、host="0.0.0.0"になるオプションを追記します。

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

python main.py を実行します。

$ python main.py 

下記のurlを実行するとFastAPIが動作していることがわかります。

(本来ならばEC2のIPv4の表記なのですが表記上、127.0.0.1とします。)

• http://127.0.0.1:8000/

  

• http://127.0.0.1:8000/items/10?q=test

  

• http://127.0.0.1:8000/docs

  

CRUD機能実装

EC2⇄RDSの接続設定

RDS操作に必要なライブラリをインストールします。

$ pip install sqlalchemy
$ pip install pymysql

EC2からのMYSQLへのトラフィックを許可するために、RDSのセキュリティグループを変更します。

データベースを作成する

RDSに新規のデータベースを生成するcreate_db.pyを作成し実行します。

import sqlalchemy
engine = sqlalchemy.create_engine('mysql+pymysql://user:password@RDS-Endpoint:3306') # サーバーと接続

dbname = 'db_fastapi' # db名
engine.execute('CREATE DATABASE %s' % dbname) #dbを作成

python create_db.py 

データベースに必要な情報まとめ

このファイルにはデータベースの設定に必要な情報をまとめたファイルdatabase.pyを作成します。

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

SQLALCHEMY_DATABASE_URI = "mysql+pymysql://user:password@RDS-Endpoint:3306/db_fastapi"

engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

class Todo(Base): # Todoテーブルの定義
    __tablename__ = 'todos'
    id = Column('id', Integer, primary_key = True)
    title = Column('title', String(200))

Base.metadata.create_all(bind=engine) # テーブル作成

database.pyを実行することでtodosテーブルが生成されます。

$ python database.py

CRUD機能を実装する

下記リンクを参考にCRUD機能を実装するようにmain.pyを修正します。

FastAPI｜DB接続してCRUDするPython製APIサーバーを構築

import uvicorn
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from pydantic import BaseModel
from database import Todo, engine, SessionLocal

app = FastAPI()

class TodoCreate(BaseModel): # データの定義
    title: str

def get_db(): # リクエスト時にSessionLocalを作成し完了したら終了する
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

@app.get("/todos/") # すべてのtodoを取得する
def read_todos(db: Session = Depends(get_db)):
    return db.query(Todo).all()

@app.get("/todos/{todo_id}")# 単一のtodoを取得する
def read_todos_only(todo_id: int, db: Session = Depends(get_db)):
    return db.query(Todo).filter(Todo.id == todo_id).first()

@app.post("/todos/") # todoを登録する
async def create_todo(todo_create: TodoCreate, db: Session = Depends(get_db)):
    todo = Todo(title=todo_create.title)
    db.add(todo)
    db.commit()
    return db.query(Todo).filter(Todo.id == todo.id).first()

@app.put("/todos/{todo_id}") # todoを更新する
async def update_todo(todo_id: int, todo_create: TodoCreate, db: Session = Depends(get_db)):
    todo = db.query(Todo).filter(Todo.id == todo_id).first()
    todo.title = todo_create.title
    db.commit()
    return db.query(Todo).filter(Todo.id == todo_id).first()

@app.delete("/todos/{todo_id}") # todoを更新する
async def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = db.query(Todo).filter(Todo.id == todo_id).first();
    db.delete(todo)
    db.commit()
    
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

main.pyファイルの修正後、docsに入り動作を確認します。

$ python main.py

• http://127.0.0.1:8000/docs　を確認するとCRUD機能が反映されていることがわかります。

  

• http://127.0.0.1:8000/docs#/default/read_todos_todos__get

  下の画像のようにgetコマンドの詳細も確認できました。

curlコマンドで確認すると全データの確認ができました。

curl -X GET "http://127.0.0.1:8000/todos/" -H "accept: application/json" 
[{"title":"test3","id":3},{"title":"test4","id":4},{"title":"test5","id":5},{"title":"sdfafff","id":6}]%   

まとめ

FastAPIに興味があり、せっかくならRDSと接続したいと思い作成しました。いろんな方の作成された記事を読ませてもらいCRUD機能までは作ることができて良かったです。 個人的には、接続時に使用しているsqlalchemyを扱うのも初めてでしたので操作に詰まりましたがpythonからデータベース操作をするにはけっこう使えるという発見もありました。FastAPIは簡易的にAPIを作成できるのでこの記事がどなたかの助けになれば幸いです。

参考リンク

• Amazon Linux 2 で Boto 3 ライブラリを使用して Python 3 仮想環境を作成する方法を教えてください。
• FastAPI
• Introduction
• howto create db mysql with sqlalchemy
• FastAPI｜DB接続してCRUDするPython製APIサーバーを構築
• sqlalchemy
• SQLAlchemyが思いのほか使えない